package com.example.bigdata.kafka.admin;


import com.typesafe.sslconfig.util.PrintlnLogger;
import org.apache.kafka.clients.admin.*;
import org.apache.kafka.common.config.ConfigResource;
import scala.actors.threadpool.Arrays;

import java.util.*;


/**
 *  设置AdminClient
 */
public class AdminSample {

    public final static String TOPIC_NAME = "kafkaStreamOut";
    public static void main(String[] args) throws Exception  {
        // 创建topic
//        createTopic();
        // 打印topic
//        topicLists();
        // 删除topic
//        delTopics();
        // 修改topic
//        alterConfig();
        // topic 描述
//        describeTopics();           //kafka监控管理使用
//        describeConfig();           //kafka监控管理使用
//        topicLists();
        // 增加partition 数量
//        incrPartitions(2);
//        describeTopics();
    }

    /**
     *  增加partition 数量
     */

    public static void incrPartitions(int partitions) throws Exception {
        AdminClient adminClient = adminClient();
        Map<String,NewPartitions> partitionsMap = new HashMap<>();
        NewPartitions newPartitions = NewPartitions.increaseTo(partitions);
        partitionsMap.put(TOPIC_NAME,newPartitions);
        CreatePartitionsResult createPartitionsResult = adminClient.createPartitions(partitionsMap);
        createPartitionsResult.all().get();
    }
        /**
         *
         *  查看 Config
         */
    public static void describeConfig() throws Exception {
        AdminClient adminClient = adminClient();
        ConfigResource configResource = new ConfigResource(ConfigResource.Type.TOPIC, TOPIC_NAME);
        DescribeConfigsResult describeConfigsResult = adminClient.describeConfigs(Arrays.asList(new ConfigResource[]{configResource}));
        Map<ConfigResource,Config> stringTopicDescriptionMap = describeConfigsResult.all().get();
        Set<Map.Entry<ConfigResource,Config>> entries = stringTopicDescriptionMap.entrySet();
        entries.stream().forEach((entry)->{
            System.out.println("ConfigResource : "+ entry.getKey()+", Config : "+ entry.getValue());
        });

    }

    /**
     * Topic 信息描述
     */
    public static void describeTopics() throws Exception {
        AdminClient adminClient = adminClient();
        DescribeTopicsResult describeTopicsResult = adminClient.describeTopics(Arrays.asList(new String[]{TOPIC_NAME}));
        Map<String,TopicDescription> stringTopicDescriptionMap = describeTopicsResult.all().get();
        Set<Map.Entry<String, TopicDescription>> entries = stringTopicDescriptionMap.entrySet();
        entries.stream().forEach((entry)->{
            System.out.println("name : "+ entry.getKey()+", desc : "+ entry.getValue());
        });

    }

    /**
     * 配置修改
     */
    public static void alterConfig() throws Exception {
        AdminClient adminClient = adminClient();
        // 2.3 以前版本API
//        Map<ConfigResource,Config> configMaps = new HashMap<>();
//        // 组织两个参数
//        ConfigResource configResource = new ConfigResource(ConfigResource.Type.TOPIC,TOPIC_NAME);
//        Config  config = new Config(Arrays.asList(new ConfigEntry[]{new ConfigEntry("preallocate", "true")}));
//        configMaps.put(configResource,config);
//        AlterConfigsResult alterConfigsResult = adminClient.alterConfigs(configMaps);

        // 从2.3以上的版本新修改的API
        // 组织两个参数
        Map<ConfigResource,Collection<AlterConfigOp>>  configMaps = new HashMap<>();
        ConfigResource configResource = new ConfigResource(ConfigResource.Type.TOPIC,TOPIC_NAME);
        AlterConfigOp alterConfigOp =
                new AlterConfigOp(new ConfigEntry("preallocate", "true"),AlterConfigOp.OpType.SET);
        configMaps.put(configResource,Arrays.asList(new AlterConfigOp[]{alterConfigOp}));
        AlterConfigsResult alterConfigsResult = adminClient.incrementalAlterConfigs(configMaps);
        alterConfigsResult.all().get();
    }

    public static void delTopics() throws Exception {
        AdminClient adminClient = adminClient();
        DeleteTopicsResult deleteTopicsResult = adminClient.deleteTopics(Arrays.asList(new String[]{TOPIC_NAME}));
        deleteTopicsResult.all().get();
    }
    /**
     * 获取Topic 列表
     */
    public static void topicLists() throws Exception {
        AdminClient adminClient = adminClient();
        ListTopicsOptions options = new ListTopicsOptions();
        // 是否查看internal 选项
        options.listInternal(true);
        ListTopicsResult listTopicsResult = adminClient.listTopics();
//        ListTopicsResult listTopicsResult = adminClient.listTopics(options);
        Set<String> names = listTopicsResult.names().get();
        // 打印names
//        names.stream().forEach(System.out::println);
        names.forEach(name -> System.out.println("Topic: " + name)); // 使用Java8的forEach方法來打印結果
        Collection<TopicListing> topicListings = listTopicsResult.listings().get();
        topicListings.stream().forEach((topicList)->{
            System.out.println(topicList+"\n");  //(name=topic1, internal=false)
        });

        // 打印
    }
    /**
     * 创建topic 实例
     */
    public static void createTopic(){
        AdminClient adminClient = adminClient();
        // 复制因子
        Short rs = 1 ;
        NewTopic newTopic = new NewTopic(TOPIC_NAME,1,rs);
        CreateTopicsResult topic = adminClient.createTopics(Arrays.asList(new NewTopic[]{newTopic}));
        System.out.println("CreateTopicsResult : " + topic);

    }
    public static AdminClient adminClient(){
        Properties properties = new Properties();
        String brokerUrl = "172.23.6.159:9092,172.23.6.160:9092,172.23.6.161:9092";
        properties.put(AdminClientConfig.BOOTSTRAP_SERVERS_CONFIG, brokerUrl);
        AdminClient adminClient = AdminClient.create(properties);
        return adminClient ;
    }
}
